package com.ruoyi.flink.util.read;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.bson.Document;

import com.alibaba.fastjson.JSONObject;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoCredential;
import com.mongodb.ServerAddress;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;

/**
 * @Description: 
 * @ClassName: MongoRead
 * @author syc
 * @date: 2022年5月30日 下午7:27:16
 */
public class MongoRead extends RichSourceFunction<JSONObject>{

	private static final long serialVersionUID = 1L;
	private  MongoClient client = null;
	private MongoCollection<Document> collection = null;
	public MongoRead() {
		
	}
	
	@Override
	public void open(Configuration parameters) throws Exception {
	    MongoClientOptions.Builder builder = new MongoClientOptions.Builder();
        builder.connectionsPerHost(10); // 每个地址最大连接数
        builder.connectTimeout(5000); // 链接超时时间 (单位毫秒)
        builder.socketTimeout(5000);  // 设置读写操作超时时间
        MongoCredential credential = MongoCredential.createCredential("root","admin","sykj85575700".toCharArray());
        ServerAddress address = new ServerAddress("192.168.1.125",27017);
        client = new MongoClient(address,credential,builder.build());
        collection = client.getDatabase("test").getCollection("test_user");
	}
	
	@Override
	public void run(SourceContext<JSONObject> ctx) throws Exception {
		 FindIterable<Document> iterable = collection.find();
		 MongoCursor<Document> cursor = iterable.iterator();
         while(cursor.hasNext()){
            // 每掉一次next移动以下指针
            Document docu = cursor.next();
            System.out.println(docu.get("name")+"\t" + docu.get("age") + "\t" + docu.get("sex") + "\t" + docu.get("secrets"));
            JSONObject json = new JSONObject();
            json.put("aaa",docu.toString());
            ctx.collect(json);
            
           // System.out.println(docu.get("name")+"\t" + docu.get("age") + "\t" + docu.get("sex") + "\t" + docu.get("secrets"));
         }
	}

	@Override
	public void cancel() {
		if(client != null) {
			client.close();
		}
		
	}

}
